更多|本文_Springboot系列(三十二):Springboot集成kafka(环境搭建+演示)|超级详细，建议收藏

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Springboot系列(三十二):Springboot集成 kafka(环境搭建+演示)|超级详细，建议收藏相关的知识，希望对你有一定的参考价值。

&＃x1f468;‍&＃x1f393; 作者&＃xff1a;bug菌

&＃x1f389;简介&＃xff1a;在CSDN、掘金等社区优质创作者&＃xff0c;全网合计6w粉&＃43;&＃xff0c;对一切技术都感兴趣&＃xff0c;重心偏java方向&＃xff0c;目前运营公众号[猿圈奇妙屋]&＃xff0c;欢迎小伙伴们的加入&＃xff0c;一起秃头。

&＃x1f6ab;特别声明&＃xff1a;原创不易&＃xff0c;转载请附上原文出处链接和本文声明&＃xff0c;谢谢配合。

&＃x1f64f;版权声明&＃xff1a;文章里可能部分文字或者图片来源于互联网或者百度百科&＃xff0c;如有侵权请联系bug菌处理。

         嗨&＃xff0c;家人们&＃xff0c;我是bug菌呀&＃xff0c;我又来啦。今天我们来聊点什么咧&＃xff0c;OK&＃xff0c;接着为大家更《springboot零基础入门教学》系列文章吧。希望能帮助更多的初学者们快速入门&＃xff01;

       小伙伴们在批阅文章的过程中如果觉得文章对您有一丝丝帮助&＃xff0c;还请别吝啬您手里的赞呀&＃xff0c;大胆的把文章
点亮&＃x1f44d;吧&＃xff0c;您的点赞三连(
收藏⭐️&＃43;关注&＃x1f468;‍&＃x1f393;&＃43;留言&＃x1f4c3;)就是对bug菌我创作道路上最好的鼓励与支持&＃x1f618;。时光不弃&＃x1f3c3;&＃x1f3fb;‍♀️&＃xff0c;创作不停&＃x1f495;&＃xff0c;加油☘️

一、前言&＃x1f525;

       上一期&＃xff0c;我是带着大家入门了springboot集成RabbitMq&＃xff0c;今天我再来一期kafka的零基础教学吧。不知道大家对kafka有多少了解&＃xff0c;反正我就是从搭建开始&＃xff0c;然后再加一个简单演示&＃xff0c;这就算是带着大家了个门哈&＃xff0c;剩下的我再后边慢慢出教程给大家说。

二、环境说明&＃x1f525;

演示环境&＃xff1a;idea2019.3 &＃43; springboot 2.3.1REALSE &＃43; windows10 &＃43; kafka

三、概念&＃x1f525;

        kafka是linkedin开源的分布式发布-订阅消息系统&＃xff0c;目前归属于Apache的顶级项目。主要特点是基于pull模式来处理消息消费&＃xff0c;追求高吞吐量&＃xff0c;在一台普通的服务器上既可以达到10W/s的吞吐速率&＃xff1b;完全的分布式系统。

        一开始的目的是日志的收集和传输。0.8版本开始支持复制&＃xff0c;不支持事务&＃xff0c;对消息的丢失&＃xff0c;重复&＃xff0c;错误没有严格要求 适用于产生大量数据的互联网服务的数据收集业务。在廉价的服务器上都能有很高的性能&＃xff0c;这个主要是基于操作系统底层的pagecache&＃xff0c;不用内存胜似使用内存。

       综上所述&＃xff0c;kafka是一款开源的消息引擎系统(消息队列/消息中间件) 分布式流处理平台。

四、Windows安装kafka&＃x1f525;

1️⃣下载kafka安装包

下载地址&＃xff1a;https://www.apache.org/dyn/closer.cgi?path&＃61;/kafka/3.1.0/kafka_2.13-3.1.0.tgz

下载完后是这么个东西&＃xff1a;

2️⃣下载好后&＃xff0c;进行解压

 3️⃣配置修改

        进入config目录&＃xff0c;修改server.properties文件,把 log.dirs的值改成 ./logs

 4️⃣kafka启动

        在你的安装目录下的bin\\windows目录上直接输入cmd然后回车。

执行命令&＃xff1a;输入

kafka-server-start.bat ../../config/server.properties

 然后可以看到控制台启动报错&＃xff1a;

        可以看到&＃xff0c;kafka还依赖于zookeeper。所以我们接下来再安装zookeeper&＃xff0c;启动zookeeper后再启动kafka试试。

五、安装zookeeper&＃x1f525;

1️⃣下载zookeeper安装包

        先下载zookeeper&＃xff0c;下载地址&＃xff1a;https://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.7.0

 2️⃣解压

        下载完后进行解压&＃xff0c;解压到指定的目录下&＃xff1b;

        然后再将apache-zookeeper-3.7.0-bin也一并下载&＃xff0c;进行解压完后&＃xff0c;将目录下的lib文件夹复制到apache-zookeeper-3.7.0主目录下&＃xff1a;

 否则后续启动肯定会报错。

 3️⃣修改配置

        修改zoo_sample.cfg 文件名(./conf) 为 zoo.cfg

        同样再编辑它指定日志位置&＃xff0c;具体配置文件如下&＃xff1a;(这里使用notepad&＃43;&＃43;小绿本进行编辑)&＃xff1b;我这里是指定在同级目录下了。这个你根据实际情况而定。

         具体添加如下&＃xff1a;仅供参考复制。

#原目录;直接注释掉
#dataDir&＃61;/tmp/zookeeper
#指定新目录
#保存数据的目录
dataDir&＃61;./data
# 保存日志文件的目录
dataLogDir&＃61;./log

4️⃣启动zookeeper

        进入到bin目录&＃xff0c;并且启动zkServer.cmd&＃xff0c;这个脚本中会启动一个java进程。

        如果你们遇到这个报错&＃xff0c;请看上边第2点zookeeper解压。报错是因为找不到类包缺少lib这个jar包文件夹&＃xff0c;所以你得下载bin包并把lib依赖都复制过来。

输入如下命名进行启动zookeeper

zkServer.cmd

具体启动如下&＃xff1a;

 正常启动截图&＃xff1a;

        再启动后jps可以看到QuorumPeerMain的进程。直接win&＃43;R 输入cmd然后再输入如下命令即可进行查询。

启动命令如下&＃xff1a;

jps -l -v

启动客户端连接一下&＃xff1a;

进入到/bin目录下&＃xff0c;执行如下命令&＃xff1a;

zkCli.cmd 127.0.0.1:2181

        如上可以看到&＃xff0c; zookeeper启动ok。至此&＃xff0c;zookeeper就安装完成啦。

        所以我们再来启动一下kafka看看&＃xff0c;是否还会跟刚才一样报错。

        切记不要关zookeeper启动服务小黑窗&＃xff0c;也就是你执行那串命令的窗口&＃xff0c;若是关了你再重新启动即可。

zkServer.cmd

kafka启动成功截图&＃xff1a;

六、kafka项目集成&＃x1f525;

1️⃣pom引入



org.springframework.kafka
spring-kafka


2️⃣配置kafka

#配置kafka 服务器
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
producer:
# 发生错误后&＃xff0c;消息重发的次数。
retries: 0
#当有多个消息需要被发送到同一个分区时&＃xff0c;生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小&＃xff0c;按照字节数计算。 batch-size: 16384 # 设置生产者内存缓冲区的大小。
buffer-memory: 33554432
# 键的序列化方式
key-serializer: org.apache.kafka.common.serialization.StringSerializer
# 值的序列化方式
value-serializer: org.apache.kafka.common.serialization.StringSerializer
# acks&＃61;0 &＃xff1a; 生产者在成功写入消息之前不会等待任何来自服务器的响应。
# acks&＃61;1 &＃xff1a; 只要集群的首领节点收到消息&＃xff0c;生产者就会收到一个来自服务器成功响应。
# acks&＃61;all &＃xff1a;只有当所有参与复制的节点全部收到消息时&＃xff0c;生产者才会收到一个来自服务器的成功响应。
acks: 1
consumer:
# 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式&＃xff0c;如1S,1M,2H,5D
auto-commit-interval: 1S
# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理&＃xff1a;
# latest&＃xff08;默认值&＃xff09;在偏移量无效的情况下&＃xff0c;消费者将从最新的记录开始读取数据&＃xff08;在消费者启动之后生成的记录&＃xff09;
# earliest &＃xff1a;在偏移量无效的情况下&＃xff0c;消费者将从起始位置读取分区的记录
auto-offset-reset: earliest
# 是否自动提交偏移量&＃xff0c;默认值是true,为了避免出现重复数据和数据丢失&＃xff0c;可以把它设置为false,然后手动提交偏移量
enable-auto-commit: false
# 键的反序列化方式
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# 值的反序列化方式
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
listener:
# 在侦听器容器中运行的线程数。
concurrency: 5
#listner负责ack&＃xff0c;每调用一次&＃xff0c;就立即commit
ack-mode: manual_immediate
missing-topics-fatal: false

3️⃣topic初始化

package com.example.demo.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* topic初始化
*
* &＃64;author luoYong
* &＃64;version 1.0
* &＃64;date 2022/2/28 17:39
*/
&＃64;Configurationpublic class KafkaConfig
/**
* 创建一个名为topic.test的Topic并设置分区数为8&＃xff0c;分区副本数为2
*/
&＃64;Bean public NewTopic initialTopic()
return new NewTopic("topic.test", 8, (short) 2);



4️⃣定义一个kafka消息发送端

package com.example.demo.component.kafka;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* kafka消息发送端
*
* &＃64;author luoYong
* &＃64;version 1.0
* &＃64;date 2022/2/28 17:40
*/
&＃64;Component
&＃64;Slf4j
public class KafkaProducer
&＃64;Autowired
private KafkaTemplate kafkaTemplate;
public void send(Object obj)
String obj2String &＃61; JSONObject.toJSONString(obj);
// 发送消息
kafkaTemplate.send("topic.test", obj).addCallback(new ListenableFutureCallback>()
&＃64;Override
public void onFailure(Throwable throwable)
// 发送失败的处理
log.info("topic[] 生产者 发送消息失败[]", "topic.test", throwable.getMessage());

&＃64;Override
public void onSuccess(SendResult stringObjectSendResult)
// 成功的处理
log.info("topic[] 生产者 发送消息成功[]", "topic.test", stringObjectSendResult.getProducerRecord().value());

);



5️⃣定义一个kafka消息消费端

package com.example.demo.component.kafka;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**
* 消息接收端-支持多端消费
*
* &＃64;author luoYong
* &＃64;version 1.0
* &＃64;date 2022/2/28 17:42
*/
&＃64;Component&＃64;Slf4jpublic class KafkaConsumer
&＃64;KafkaListener(topics &＃61; "topic.test", groupId &＃61; "topic.group1")
public void topicTest(ConsumerRecord record, Acknowledgment ack, &＃64;Header(KafkaHeaders.RECEIVED_TOPIC) String topic)
Optional message &＃61; Optional.ofNullable(record.value());
if (message.isPresent())
Object msg &＃61; message.get();
log.info("客户端 A 消费了&＃xff1a; Topic[] Message[]", topic, msg);
ack.acknowledge();


&＃64;KafkaListener(topics &＃61; "topic.test", groupId &＃61; "topic.group1")
public void topicTest1(ConsumerRecord record, Acknowledgment ack, &＃64;Header(KafkaHeaders.RECEIVED_TOPIC) String topic)
Optional message &＃61; Optional.ofNullable(record.value());
if (message.isPresent())
Object msg &＃61; message.get();
log.info("客户端 B 消费了&＃xff1a; Topic[] Message[]", topic, msg);
ack.acknowledge();




6️⃣定义一个测试类进行测试

/**
* &＃64;author luoYong
* &＃64;version 1.0
* &＃64;date 2022/2/24 17:02
*/
&＃64;RunWith(SpringRunner.class)
&＃64;SpringBootTest(webEnvironment &＃61; SpringBootTest.WebEnvironment.RANDOM_PORT)
public class KafkaTest
&＃64;Autowired
private KafkaProducer kafkaProducer;
&＃64;Test
public void testSendMsg()
String msg &＃61; "hello";
kafkaProducer.send(msg);



7️⃣测试结果如下

       ... ...

       ok&＃xff0c;以上就是我这期的全部内容啦&＃xff0c;如果还想学习更多&＃xff0c;你可以看看我的往期热文推荐哦&＃xff0c;每天积累一个奇淫小知识&＃xff0c;日积月累下去&＃xff0c;你一定能成为令人敬仰的大佬的。好啦&＃xff0c;咱们下期见~

七、往期推荐&＃x1f525;

• springboot系列(一):如何创建springboot项目及启动

• springboot系列(二):yaml、properties两配置文件介绍及使用

• springboot系列(三):多环境切换&＃xff0c;实例演示
• springboot系列(四):stater入门
• springboot系列(五):史上最最最全springboot常用注解
• springboot系列(六):mysql配置及数据库查询
• springboot系列(七):如何通过mybatis-plus实现接口增删改查
• springboot系列(八):mybatis-plus之条件构造器使用手册
• springboot系列(九):mybatis-plus之如何自定义sql
• springboot系列(十):mybatis之xml映射文件&＃xff1e;、&＃xff1c;&＃61;等特殊符号写法
• springboot系列(十一):实现多数据源配置&＃xff0c;开箱即用
• springboot系列(十二):如何实现邮件发送提醒&＃xff0c;你一定得会(准备篇)
• springboot系列(十三):如何实现发送普通邮件&＃xff1f;你一定得会
• springboot系列(十四):如何实现发送图片、doc文档等附件邮件&＃xff1f;你一定得会
• springboot系列(十五):如何实现静态邮件模板发送&＃xff1f;你一定得会
• springboot系列(十六):如何实现发送邮件提醒&＃xff0c;附完整源码
• springboot系列(十七):集成在线接口文档Swagger2
• springboot系列(十八):如何Windows安装redis&＃xff1f;你玩过么
• springboot系列(十九):如何集成redis&＃xff1f;不会我教你
• springboot系列(二十):如何通过redis实现手机号验证码功能 
• ... ...

八、文末&＃x1f525;

        如果还想要学习更多&＃xff0c;小伙伴们可关注bug菌专门为大家创建的专栏《springboot零基础入门教学》&＃xff0c;从无到有&＃xff0c;从零到一&＃xff01;希望能帮助到更多小伙伴们。

       我是bug菌&＃xff0c;一名想走&＃x1f463;出大山改变命运的程序猿。接下来的路还很长&＃xff0c;都等待着我们去突破、去挑战。来吧&＃xff0c;小伙伴们&＃xff0c;我们一起加油&＃xff01;未来皆可期&＃xff0c;fighting&＃xff01;

        最后送大家两句我很喜欢的话&＃xff0c;与诸君共勉&＃xff01;

☘️做你想做的人&＃xff0c;没有时间限制&＃xff0c;只要愿意&＃xff0c;什么时候都可以start。

&＃x1f340;你能从现在开始改变&＃xff0c;也可以一成不变&＃xff0c;这件事&＃xff0c;没有规矩可言&＃xff0c;你可以活出最精彩的自己。

​​​

&＃x1f48c;如果文章对您有所帮助&＃xff0c;就请留下您的赞吧&＃xff01;(#^.^#)&＃xff1b;

&＃x1f49d;如果喜欢bug菌分享的文章&＃xff0c;就请给bug菌点个关注吧&＃xff01;(๑′ᴗ‵๑)づ╭❤&＃xff5e;&＃xff1b;

&＃x1f497;如果对文章有任何疑问&＃xff0c;还请文末留言或者加群吧&＃xff1b;

&＃x1f49e;鉴于个人经验有限&＃xff0c;所有观点及技术研点&＃xff0c;如有异议&＃xff0c;请直接回复参与讨论(请勿发表攻击言论&＃xff0c;谢谢&＃xff09;&＃xff1b;

&＃x1f495;版权声明&＃xff1a;原创不易&＃xff0c;转载请附上原文出处链接和本文声明&＃xff0c;版权所有&＃xff0c;盗版必究&＃xff01;&＃xff01;&＃xff01;谢谢。